/*
 * Copyright (c) 2023. All rights reserved.
 * Please don't try to modify the code, because I don't know what the consequences will be.
 *
 *
 *
 *
 *
 *
 *
 */

package cc.zzone.bean;

import cc.zzone.client.BiliWsClient;
import cc.zzone.client.DouyuWsClient;
import cc.zzone.domain.ChatConf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.springframework.util.Assert;
import org.springframework.util.MultiValueMap;
import org.springframework.util.StringUtils;
import org.springframework.web.util.UriComponents;
import org.springframework.web.util.UriComponentsBuilder;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * websocket server
 * 当有客户端连接时，按照confId开启该平台房间的弹幕监听服务（同平台房间仅开启一个监听服务）
 * 当有客户端离线且该平台房间客户端已为空，则关闭该弹幕监听服务
 *
 * @author jinzhengze
 * @since 2023年5月10日
 */
@Slf4j
public final class ChatWebsocketServer extends ServerInfo {

    public static final Map<String, ChannelGroup> CHANNEL_MAP = new ConcurrentHashMap<>(); // 当前连接客户端
    public static final Map<String, WebSocketClient> WS_MAP = new ConcurrentHashMap<>(); // 当前开启的监听服务

    private final String path;

    public ChatWebsocketServer(String path) {
        Assert.notNull(path, "path can not be null");
        this.path = path;
    }

    @Override
    public List<ChannelHandler> getHandlers() {
        List<ChannelHandler> handlers = new ArrayList<>();
        handlers.add(new HttpServerCodec()); // 基于 http 协议，使用 http 的编码和解码器
        handlers.add(new HttpObjectAggregator(65535));
        handlers.add(onlineHandler());
        handlers.add(new ChunkedWriteHandler());
        handlers.add(new WebSocketServerProtocolHandler(path));
        handlers.add(encodeHandler());
        handlers.add(decodeHandler());
        handlers.add(websocketHandler());
        return handlers;
    }

    private ChannelHandler onlineHandler() {
        return new SimpleChannelInboundHandler<FullHttpRequest>() {
            @Override
            protected void channelRead0(ChannelHandlerContext context,
                                        FullHttpRequest fullHttpRequest) {
                UriComponents uriComponents = UriComponentsBuilder.fromUriString(fullHttpRequest.uri()).build();
                MultiValueMap<String, String> queryParams = uriComponents.getQueryParams();
                String confId = queryParams.getFirst("confId");
                if (!StringUtils.hasText(confId)) {
                    context.close();
                    return;
                }
                fullHttpRequest.headers().set("confId", confId);
                fullHttpRequest.setUri(path);
                context.fireChannelRead(fullHttpRequest.retain());
            }
        };
    }

    private ChannelHandler encodeHandler() {
        return new MessageToMessageEncoder<String>() {
            @Override
            protected void encode(ChannelHandlerContext context,
                                  String msg,
                                  List<Object> list) {
                list.add(new TextWebSocketFrame(msg));
            }
        };
    }

    private ChannelHandler decodeHandler() {
        return new MessageToMessageDecoder<WebSocketFrame>() {
            @Override
            protected void decode(ChannelHandlerContext channelHandlerContext,
                                  WebSocketFrame webSocketFrame,
                                  List<Object> list) {
                if (webSocketFrame instanceof TextWebSocketFrame tframe) {
                    // 内存泄漏
                    list.add(tframe.text());
                } else {
                    // 不必要，但这种情况下 release 无害
                    webSocketFrame.release();
                }
            }
        };
    }

    private ChannelHandler websocketHandler() {
        return new SimpleChannelInboundHandler<>() {
            @Override
            public void userEventTriggered(ChannelHandlerContext context, Object evt) {
                if (evt instanceof IdleStateEvent) {
                    // 心跳超时日志
                    heartTimeoutLog(context);
                    context.close();
                } else if (evt instanceof WebSocketServerProtocolHandler.HandshakeComplete complete) {
                    // 终端上线日志
                    onlineLog(context);
                    // 终端上线操作
                    Integer confId = Integer.parseInt(complete.requestHeaders().get("confId"));
                    ChatConf conf = ChatConf.mapper().getData(confId);
                    if (conf == null) {
                        log.error("【弹幕配置不存在，confId={}】", confId);
                        context.close();
                        return;
                    }
                    ChannelGroup channelGroup = CHANNEL_MAP.computeIfAbsent(conf.getKey(),
                            k -> new DefaultChannelGroup(GlobalEventExecutor.INSTANCE));
                    channelGroup.add(context.channel());
                    // 客户端连接数为1且服务监听不存在
                    if ((WS_MAP.get(conf.getKey()) == null || WS_MAP.get(conf.getKey()).isClosed()) &&
                            channelGroup.size() == 1) {
                        try { // 开启弹幕服务监听
                            WebSocketClient client = null;
                            if ("BILI".equals(conf.getPlatform())) client = BiliWsClient.getInstance(conf);
                            else if ("DY".equals(conf.getPlatform())) client = DouyuWsClient.getInstance(conf);

                            if (client != null) {
                                client.connect();
                                WS_MAP.put(conf.getKey(), client);
                            }
                        } catch (Exception e) {
                            log.error("【弹幕服务监听器启动失败】", e);
                            channelGroup.remove(context.channel());
                            context.close();
                        }
                    }
                    List<String> result = new ArrayList<>();
                    for (String key : CHANNEL_MAP.keySet())
                        if (CHANNEL_MAP.get(key) != null && CHANNEL_MAP.get(key).size() > 0)
                            result.add(key + ":" + CHANNEL_MAP.get(key).size());
                    log.info("【当前转发服务】{}", result);
                }
            }

            @Override
            protected void channelRead0(ChannelHandlerContext context, Object msg) {
                // 客户端信息日志
                readLog(context, msg);
            }

            @Override
            public void handlerRemoved(ChannelHandlerContext context) {
                // 下线日志
                removedLog(context);
                List<String> result = new ArrayList<>();
                // 若平台房间的客户端在线数为0，则关闭该转发服务的监听
                for (Map.Entry<String, ChannelGroup> entry : CHANNEL_MAP.entrySet()) {
                    ChannelGroup channelGroup = entry.getValue();
                    channelGroup.remove(context.channel());
                    if (channelGroup.size() == 0) {
                        WebSocketClient webSocketClient = WS_MAP.get(entry.getKey());
                        if (webSocketClient != null && webSocketClient.isOpen())
                            webSocketClient.close();
                        continue;
                    }
                    // 记录在线转发服务
                    result.add(entry.getKey() + ":" + channelGroup.size());
                }
                log.info("【当前转发服务】{}", result);
            }

            @Override
            public void exceptionCaught(ChannelHandlerContext context, Throwable cause) {
                // 异常日志
                exceptionLog(context, cause);
                // 断开连接
                context.close();
            }
        };
    }
}
